#include "SysDefine.hpp"
#include "MqttClient.hpp"

const string MQ_DEV_TYPE = "dtu";

// 连接成功回调
static void on_connect(struct mosquitto* mosq, void* obj, int reason_code)
{
	cout << "on_connect: " << mosquitto_connack_string(reason_code) << endl;
	if (reason_code != 0) {
		mosquitto_disconnect(mosq);
    return;
	}

	// 订阅cmd主题
	// int res = mosquitto_subscribe(mosq, NULL, S_GETP("topic-cmd"), 0);
	// if (res != MOSQ_ERR_SUCCESS) {
	// 	cout << "subscribing err:" << mosquitto_strerror(res) << "\n";
	// 	mosquitto_disconnect(mosq);
  //   return;
	// }
	// cout << " - sub topic '" << S_GETP("topic-cmd") << "' ok" << endl;

	// res = mosquitto_subscribe(mosq, NULL, S_GETP("topic-tx"), 0);
	// if (res != MOSQ_ERR_SUCCESS) {
	// 	cout << "subscribing err:" << mosquitto_strerror(res) << "\n";
	// 	mosquitto_disconnect(mosq);
  //   return;
	// }
	// cout << " - sub topic '" << S_GETP("topic-tx") << "' ok" << endl;
}

// 连接断开回调
static void on_disconnect(struct mosquitto* mosq, void* obj, int reason_code)
{
	cout << "on_disconnect: " << mosquitto_connack_string(reason_code) << endl;
	zmq::socket_t *pPubSock = static_cast<zmq::socket_t *>(obj); // obj是mosquitto_new传入的参数
	pPubSock->close();
}

// 订阅会包回调
static void on_subscribe(struct mosquitto* mosq, void* obj, int mid, int qos_count, const int* granted_qos)
{
	bool subIsOk = false;
	for (int i = 0; i < qos_count; i++) {
		cout << "on_subscribe: ret=" << i << ", qos=" << granted_qos[i] << endl;
		if (granted_qos[i] <= 2){
			subIsOk = true;
		}
	}
	if (subIsOk == false) {
		cout << "[ERROR] All subscriptions rejected." << endl;
		mosquitto_disconnect(mosq);
	}
}

// 收到mqtt消息回调
static void on_message(struct mosquitto* mosq, void* obj, const struct mosquitto_message* msg)
{
	string topic(msg->topic);

	// if (topic == S_GETS("topic-cmd")) { // cmd消息
	// 	string payload((char*)msg->payload);
	// 	cout << topic << ": " << payload << endl;

	// 	string outStr;
	// 	uint retLen = ExecuteShell(payload, outStr); // 执行系统命令
	// 	if (retLen == 0) return;

	// 	// 回包
	// 	int rc = mosquitto_publish(mosq, NULL, S_GETP("topic-cmdres"), retLen, outStr.c_str(), 0, false);
	// 	if (rc != MOSQ_ERR_SUCCESS) {
	// 		cout << "[ERROR] pub: " << mosquitto_strerror(rc) << endl;
	// 	}
	// } else if (topic == S_GETS("topic-tx")) { // tx消息
	// 	cout << topic << ": " << std::hex << std::showbase;
	// 	for (int i = 0; i < msg->payloadlen; i++) cout << (int)((char *)msg->payload)[i] << " ";
	// 	cout << std::dec << endl;

	// 	zmq::socket_t *pPubSock = static_cast<zmq::socket_t *>(obj); // obj是mosquitto_new传入的参数
	// 	pPubSock->send(zmq::str_buffer(ZMQ_TOPIC_US_TX), zmq::send_flags::sndmore);
	// 	pPubSock->send(zmq::const_buffer(msg->payload, msg->payloadlen), zmq::send_flags::none);
	// }
}

// 连接mq服务器
int MqttClient::ConnectMqttSrv(zmq::socket_t *pPubSock)
{
	if (S_NOT_HAS("dev-id") || S_NOT_HAS("dev-key")) {
		cout << "Not Find Dev Id & Key For Mqtt!" << endl;
		return -1;
	}

	if (S_NOT_HAS("mqtt-srv")) S_SET("mqtt-srv", "srv.12dsj.com"); // 默认服务器
	if (S_NOT_HAS("mqtt-port")) S_SET("mqtt-port", 8883);

	// 初始化配置项
	string devId = S_GETS("dev-id");
	string devKey = S_GETS("dev-key");
	S_SET("mqtt-user", MQ_DEV_TYPE);
	S_SET("mqtt-pw", devKey);
	S_SET("mqtt-id", StrJoin({MQ_DEV_TYPE, devId}, "."));
	S_SET("topic-cmd", StrJoin({MQ_DEV_TYPE, "cmd", devId}, "/"));
	S_SET("topic-cmdres", StrJoin({MQ_DEV_TYPE, "cmdres", devId}, "/"));
	S_SET("topic-tx", StrJoin({"0", MQ_DEV_TYPE, "tx", devId}, "/"));
	S_SET("topic-rx", StrJoin({"0", MQ_DEV_TYPE, "rx", devId}, "/"));
	S_SET("topic-var", StrJoin({"0", MQ_DEV_TYPE, "var", "0-"+devId}, "/"));

	// 初始化mqtt
	mosquitto_lib_init();
	_myClient = mosquitto_new(S_GETP("mqtt-id"), true, pPubSock);
	if (_myClient == NULL) {
		cout << "[ERROR] Out of memory." << endl;
		return -2;
	}
	mosquitto_username_pw_set(_myClient, S_GETP("mqtt-user"), S_GETP("mqtt-pw"));
	mosquitto_connect_callback_set(_myClient, on_connect);
	mosquitto_disconnect_callback_set(_myClient, on_disconnect);
	mosquitto_subscribe_callback_set(_myClient, on_subscribe);
	mosquitto_message_callback_set(_myClient, on_message);

	int rc = mosquitto_connect(_myClient, S_GETP("mqtt-srv"), S_GETN("mqtt-port"), 60);
	if (rc != MOSQ_ERR_SUCCESS) {
		mosquitto_destroy(_myClient);
		cout << "[ERROR] mosq:" << mosquitto_strerror(rc) << endl;
		return -3;
	}

	// 进入消息循环
	mosquitto_loop_start(_myClient); // 新开线程非阻塞方式
	// mosquitto_loop_forever(myClient, -1, 1); // 同步阻塞方式
	// mosquitto_lib_cleanup(); // 永不调用

	return 0;
}

// 发包
int MqttClient::Publish(string point, const char* payload, uint len)
{
	const char* topic;
	if (point == "file") topic = S_GETP("topic-file");
  else if (point == "file-opt") topic = S_GETP("topic-file-opt");
	else return MOSQ_ERR_NOT_SUPPORTED;

	int res = mosquitto_publish(_myClient, NULL, topic, len, payload, 0, false);
	if (res != MOSQ_ERR_SUCCESS) {
		cout << "[ERROR] pub: " << mosquitto_strerror(res) << endl;
	}
	return res;
}

// 发包，带子路径
int MqttClient::Publish(string point, string subName, const char* payload, uint len)
{
	string topic;
	if (point == "file") topic = S_GETS("topic-file") + "/" + subName;
  else if (point == "file-opt") topic = S_GETS("topic-file-opt") + "/" + subName;
	else return MOSQ_ERR_NOT_SUPPORTED;

	int res = mosquitto_publish(_myClient, NULL, topic.c_str(), len, payload, 0, false);
	if (res != MOSQ_ERR_SUCCESS) {
		cout << "[ERROR] pub: " << mosquitto_strerror(res) << endl;
	}
	return res;
}

// 发字符串包
int MqttClient::Publish(string point, string subName, const char* stringOut)
{
	return Publish(point, subName, stringOut, strlen(stringOut));
}